package com.hong.qrpc;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.hong.qrpc.annotation.QRpcClient;
import com.hong.qrpc.codec.Decoder;
import com.hong.qrpc.codec.Encoder;
import com.hong.qrpc.codec.constant.CodecConst;
import com.hong.qrpc.config.ConfigManager;
import com.hong.qrpc.config.RegisterConfig;
import com.hong.qrpc.config.RpcClientConfig;
import com.hong.qrpc.exception.transport.DisconnectException;
import com.hong.qrpc.utils.ReflectionUtils;
import com.hong.qrpc.utils.StringUtils;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
import lombok.ToString;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.IOUtils;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;

/**
 * 调用远程服务的代理类
 *
 * @Author HQT
 * @Date 2022/7/10 19:46
 */
@Slf4j
@Data
@NoArgsConstructor
@ToString
@EqualsAndHashCode
public class RemoteInvoke implements InvocationHandler {
    private Class interfaceClass;
    private Encoder encoder;
    private Decoder decoder;
    private TransportSelector selector;
    private Long timeout = 2000L;
    /**
     * 重试次数
     */
    private Integer retries = 0;
    private int serializationType;
    private RpcClient rpcClient;

    /**
     * @param encoder
     * @param decoder
     * @param selector
     */
    public RemoteInvoke(Class interfaceClass, Encoder encoder, Decoder decoder, TransportSelector selector) {
        this(interfaceClass);
        this.encoder = encoder;
        this.decoder = decoder;
        this.selector = selector;
        this.serializationType = CodecConst.getSerializationTypeBy(encoder);
    }

    public RemoteInvoke(Class interfaceClass) {
        this.interfaceClass = interfaceClass;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //如果不是接口方法的话就不进行远程调用
        switch (method.getName()) {
            case "toString":
                return toString();
            case "equals":
                return method.invoke(this, args);
            case "hashCode":
                return hashCode();
        }
        //wait for selector init
        if (selector == null) {
            log.warn("The class {} of the method cannot be called temporarily,because the initialization is not complete", interfaceClass);
            return null;
        }
        Request request = new Request();
        request.setServiceDescriptor(ServiceDescriptor.from(interfaceClass, method));
        request.setParameters(args);
        Response resp = null;
        int retries = this.retries;
        do {
            try {
                resp = invokeRemote(request);
                if (!(resp == null || resp.getStatusCode() != 0)) {
                    break;
                }
            } catch (Exception e) {
                if (retries > 0) {
                    if (resp == null) {
                        log.warn("fail to invoke remote cause:The server is unresponsive retrying now");
                    } else {
                        log.warn("fail to invoke remote   error message:{}  retrying now", resp.getMessage());
                    }
                }
            }
        } while (retries-- > 0);
        QRpcClient annotation = (QRpcClient) interfaceClass.getAnnotation(QRpcClient.class);
        Class fallbackClass = annotation.fallback();
        if (Object.class.equals(fallbackClass)) {
            if (resp == null) {
                throw new IllegalStateException("fail to invoke remote cause:The server is unresponsive");
            } else if (resp.getStatusCode() != 0) {
                throw new IllegalStateException("fail to invoke remote   error message:" + resp.getMessage());
            }
        } else {
            if (resp == null) {
                log.warn("fail to invoke remote cause:The server is unresponsive. Trigger method downgrade");
            } else if (resp.getStatusCode() != 0) {
                log.warn("fail to invoke remote and Trigger method downgrade  error message:{}", resp.getMessage());
            }
            //For performance reasons, it is not checked here whether the fallback class implements the interface
            return method.invoke(ReflectionUtils.newInstance(fallbackClass), args);
        }
        return resp.getData();
    }

    private Response invokeRemote(Request request) {
        TransportClient client = null;
        Response response = null;
        TransportClient helpRemove = null;
        try {
            client = selector.select();
            helpRemove = client;
            byte[] outBytes = encoder.encode(request);
            InputStream receive = client.write(new ByteArrayInputStream(outBytes), serializationType, timeout);
            if (receive == null) {
                //多半是请求超时了
                return null;
            }
            byte[] inBytes = IOUtils.readFully(receive, receive.available());
            response = decoder.decode(inBytes, Response.class);
        } catch (Exception e) {
            //代表服务器的连接已经断开了，需要去注册中心重新拉取服务提供者列表，不过首先要踢出原来的客户端
            if (e instanceof DisconnectException) {
                selector.release(helpRemove);
                QRpcClient annotation = (QRpcClient) interfaceClass.getAnnotation(QRpcClient.class);
                String serviceName = annotation.serviceName();
                ConfigManager configManager = rpcClient.getConfigManager();
                RegisterConfig registerConfig = configManager.getRegisterConfig();
                String registerAddrs = registerConfig.getRegisterAddrs();
                RpcClientConfig rpcClientConfig = configManager.getRpcClientConfig();
                if (!StringUtils.isEmpty(registerAddrs)) {
                    List<TransportClient> transportClients = new ArrayList<>();
                    List<Peer> helpDistinct = new ArrayList<>();
                    for (String addr : StringUtils.splitByChars(registerAddrs, ';')) {
                        try {
                            NamingService service = NamingFactory.createNamingService(addr);
                            //过滤不健康实例
                            List<Instance> instances = service.selectInstances(serviceName, true);
                            for (Instance instance : instances) {
                                Peer peer = new Peer(rpcClientConfig.getProtocolType(), instance.getIp(), instance.getPort());
                                TransportClient transportClient = ReflectionUtils.newInstance(rpcClientConfig.getTransportClass());
                                helpDistinct.add(peer);
                                transportClients.add(transportClient);
                            }
                        } catch (NacosException nacosException) {
                            nacosException.printStackTrace();
                            log.warn("fail to find register center using addr:{}", addr);
                        }
                    }
                    log.info("updating TransportClients");
                    selector.updateTransportClients(transportClients, helpDistinct);
                }
            }
            log.warn(e.getMessage(), e);
            response = new Response();
            response.setStatusCode(-1);
            response.setMessage("RpcClient got error:" + e.getClass().getName()
                    + " : " + e.getMessage());
            throw new RuntimeException(e);
        }
        return response;
    }

    /**
     * 从高优先级到低优先级进行加载配置，加载到了就返回
     *
     * @param rpcClient
     */
    public void setTimeout(RpcClient rpcClient) {
        QRpcClient annotation = (QRpcClient) interfaceClass.getAnnotation(QRpcClient.class);
        if (Long.MAX_VALUE != annotation.timeout()) {
            this.timeout = annotation.timeout();
            return;
        }
        Long timeout = rpcClient.getConfigManager().getInvokerConfig().getNumberOfTimeout();
        if (timeout != null) {
            this.timeout = timeout;
            return;
        }
    }

    public void setRetries(RpcClient rpcClient) {
        QRpcClient annotation = (QRpcClient) interfaceClass.getAnnotation(QRpcClient.class);
        if (annotation.retries() != 0) {
            this.retries = annotation.retries();
            return;
        }
        Integer retries = rpcClient.getConfigManager().getInvokerConfig().getRetries();
        if (retries != null) {
            this.retries = retries;
            return;
        }
    }
}
